package com.niit;

import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.clients.producer.ProducerRecord;
import java.text.SimpleDateFormat;
import java.util.*;

public class KafkaProducer {
    //测试git
    private static final Random random = new Random();
    private static final Map<String, List<String>> CATEGORY_PRODUCTS = new HashMap<>();
    private static final String[] CATEGORIES = {
            "电子产品", "服装鞋帽", "食品饮料", "家居用品", "美妆个护",
            "运动户外", "母婴用品", "图书音像", "玩具礼品", "生鲜水产"
    };

    static {
        // 初始化类别与商品的映射关系，增加数据多样性
        CATEGORY_PRODUCTS.put("电子产品", Arrays.asList(
                "智能手机", "笔记本电脑", "平板", "智能手表", "耳机",
                "相机", "充电器", "移动硬盘", "鼠标", "键盘"
        ));
        CATEGORY_PRODUCTS.put("服装鞋帽", Arrays.asList(
                "男士T恤", "女士连衣裙", "牛仔裤", "运动鞋", "休闲鞋",
                "羽绒服", "毛衣", "衬衫", "童装", "袜子"
        ));
        CATEGORY_PRODUCTS.put("食品饮料", Arrays.asList(
                "巧克力", "饼干", "方便面", "碳酸饮料", "果汁",
                "牛奶", "坚果", "罐头", "咖啡", "茶叶"
        ));
        CATEGORY_PRODUCTS.put("家居用品", Arrays.asList(
                "沙发", "餐桌椅", "床品四件套", "电饭煲", "吸尘器",
                "加湿器", "台灯", "餐具", "收纳盒", "拖把"
        ));
        CATEGORY_PRODUCTS.put("美妆个护", Arrays.asList(
                "口红", "粉底液", "面膜", "洗发水", "沐浴露",
                "香水", "护手霜", "剃须刀", "牙膏", "护肤品套装"
        ));
        CATEGORY_PRODUCTS.put("运动户外", Arrays.asList(
                "篮球", "足球", "瑜伽垫", "登山包", "帐篷",
                "自行车", "跑步机", "羽毛球拍", "游泳镜", "滑雪板"
        ));
        CATEGORY_PRODUCTS.put("母婴用品", Arrays.asList(
                "婴儿奶粉", "纸尿裤", "婴儿车", "奶瓶", "婴儿床",
                "婴儿服装", "孕妇装", "早教玩具", "吸奶器", "婴儿湿巾"
        ));
        CATEGORY_PRODUCTS.put("图书音像", Arrays.asList(
                "小说", "教材", "漫画", "音乐CD", "电影DVD",
                "杂志", "字帖", "百科全书", "有声读物", "绘画工具"
        ));
        CATEGORY_PRODUCTS.put("玩具礼品", Arrays.asList(
                "乐高积木", "遥控汽车", "毛绒玩具", "益智玩具", "节日礼品",
                "创意礼品", "手工DIY", "拼图", "模型", "文具"
        ));
        CATEGORY_PRODUCTS.put("生鲜水产", Arrays.asList(
                "苹果", "香蕉", "草莓", "猪肉", "牛肉",
                "鱼类", "虾", "螃蟹", "鸡蛋", "蔬菜礼盒"
        ));
    }

    public static void main(String[] args) {
        // Kafka配置
        Properties props = new Properties();
        props.put("bootstrap.servers", "192.168.136.128:9092");
        props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");

        // 创建Kafka生产者
        try (Producer<String, String> producer = new org.apache.kafka.clients.producer.KafkaProducer<>(props)) {
            while (true) {
                // 生成订单类别
                String category = getRandomCategory();

                // 生成订单名字（基于选定的类别）
                String product = getRandomProductByCategory(category);

                // 生成订单数量（根据商品类型设置不同范围）
                int quantity = getRandomQuantity(category);

                // 生成日期（随机生成近30天内的日期，增加数据多样性）
                String date = getRandomDate();

                // 生成用户ID（格式：2025XX，XX为0-9的随机数字）
                String userId = generateUserId();

                // 生成顾客评分（正态分布，均值75，标准差15）
                int rating = getNormalDistributedRating();

                // 生成是否有效（90%概率为有效订单）
                char isValid = random.nextDouble() < 0.9? 'Y' : 'N';

                // 拼接数据（用户ID添加在日期和评分之间）
                String data = String.format("%s\t%s\t%d\t%s\t%s\t%d\t%c",
                        category, product, quantity, date, userId, rating, isValid);

                // 发送数据到Kafka主题
                producer.send(new ProducerRecord<>("orders", data));
                System.out.println("Sent: " + data);

                // 控制发送频率（修改为10秒发送一次）
                Thread.sleep(10000);
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    private static String getRandomCategory() {
        return CATEGORIES[random.nextInt(CATEGORIES.length)];
    }

    private static String getRandomProductByCategory(String category) {
        List<String> products = CATEGORY_PRODUCTS.getOrDefault(category, Collections.singletonList("未知商品"));
        return products.get(random.nextInt(products.size()));
    }

    private static int getRandomQuantity(String category) {
        // 根据商品类别设置不同的数量范围
        switch (category) {
            case "生鲜水产":
            case "食品饮料":
                return 1 + random.nextInt(20);  // 1-20
            case "服装鞋帽":
            case "美妆个护":
                return 1 + random.nextInt(10);  // 1-10
            case "电子产品":
            case "家居用品":
                return 1 + random.nextInt(5);   // 1-5
            default:
                return 1 + random.nextInt(15);  // 1-15
        }
    }

    private static String getRandomDate() {
        // 生成近30天内的随机日期
        Calendar calendar = Calendar.getInstance();
        calendar.add(Calendar.DAY_OF_YEAR, -random.nextInt(30));
        return new SimpleDateFormat("yyyy-MM-dd").format(calendar.getTime());
    }

    private static int getNormalDistributedRating() {
        // 使用正态分布生成更真实的评分（大部分在70-90之间）
        double mean = 7.5;
        double stdDev = 1.5;
        int rating = (int) Math.round(mean + random.nextGaussian() * stdDev);
        return Math.max(0, Math.min(10, rating));  // 限制在0-10范围内
    }

    // 生成用户ID的方法
    private static String generateUserId() {
        // 生成2025开头，后面两位为0-9随机数字的6位用户ID
        int digit1 = random.nextInt(5);  // 0-4
        int digit2 = random.nextInt(5);  // 0-4
        return String.format("2025%d%d", digit1, digit2);
    }
}